package com.myorg.service;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.http.HttpResponse;
import org.apache.http.NameValuePair;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.message.BasicNameValuePair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.myorg.domain.BlockInfo;
import com.myorg.domain.SearchRequest;
import com.myorg.util.ConstantsToBeExternalized;
import com.myorg.util.DataLocalityRunner;
import com.myorg.util.DataLocalityRunner.DataLocalityTask;
import com.myorg.util.HdfsHelper;
import com.myorg.util.Utility;

//TODO: Auto-wiring is not there yet.
public class ScanService {
	private static final Logger logger = LoggerFactory.getLogger(ScanService.class);
	protected static ExecutorService pool = Executors
			.newFixedThreadPool(ConstantsToBeExternalized.NUM_OF_SEARCH_THREAD_ON_NODE);

	protected void dumpBlockInfo(List<BlockInfo> blkInfos) throws IOException {
		for (int i = 0; i < blkInfos.size(); i++) {
			String[] hosts = blkInfos.get(i).getBlockLocation().getHosts();
			String[] pathInfo = blkInfos.get(i).getBlockLocation().getTopologyPaths();
			for (int j = 0; j < pathInfo.length; j++) {
				logger.info("getTopologyPaths[" + i + "]" + pathInfo[j].toString());
			}
			logger.info("Host %d: %s %n", i, hosts);
		}
	}

	protected List<BlockInfo> generateBlockInfoList(String[] filePaths) throws IOException {
		HdfsHelper helper = new HdfsHelper();
		List<BlockInfo> rtn = new ArrayList<BlockInfo>();
		for (String path : filePaths) {
			rtn.addAll(helper.getBlockInfos(path));
		}
		return rtn;
	}

	public void dispatchScan(final SearchRequest request, final String contextPath, final int port,
			final String syncPath) {
		RequestConfig requestConfig = RequestConfig.custom()
				.setSocketTimeout(ConstantsToBeExternalized.DEFAULT_SOCKET_TIMEOUT)
				.setConnectTimeout(ConstantsToBeExternalized.DEFAULT_CONNECTION_TIMEOUT).build();
		final CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setDefaultRequestConfig(requestConfig)
				.setMaxConnTotal(Integer.MAX_VALUE).setMaxConnPerRoute(Integer.MAX_VALUE).build();
		try {
			List<BlockInfo> blockInfos = generateBlockInfoList(request.getFilePaths());
			dumpBlockInfo(blockInfos);

			httpclient.start();
			final CountDownLatch latch = new CountDownLatch(blockInfos.size());
			DataLocalityRunner.runWithLocality(blockInfos, new DataLocalityTask() {
				@Override
				public void run(BlockInfo blkInfo, String selectedHost) {
					try {
						URI uri = new URI("http", null, selectedHost, port, contextPath + "/DoScan",
								"start="
										+ blkInfo.getBlockLocation().getOffset()
										+ "&end="
										+ (blkInfo.getBlockLocation().getLength()
												+ blkInfo.getBlockLocation().getOffset() - 1), null);
						logger.info("Dispatching to URL: " + uri);

						final HttpPost httpPost = new HttpPost(uri);
						List<NameValuePair> params = new ArrayList<NameValuePair>(2);
						params.add(new BasicNameValuePair("source", blkInfo.getFilePath()));
						httpPost.setEntity(new UrlEncodedFormEntity(params, "UTF-8"));

						httpclient.execute(httpPost, new FutureCallback<HttpResponse>() {
							public void completed(final HttpResponse resp) {
								// TODO handle socket time out.
								logger.info(httpPost.getRequestLine() + "->" + resp.getStatusLine());
								latch.countDown();
							}

							public void failed(final Exception ex) {
								latch.countDown();
								// TODO handle socket timeout here, maybe retry or send the request to another host.
								logger.info(httpPost.getRequestLine() + "->" + ex);
							}

							public void cancelled() {
								latch.countDown();
								logger.info(httpPost.getRequestLine() + " cancelled");
							}
						});
					} catch (URISyntaxException e) {
						e.printStackTrace();
						throw new RuntimeException(e);
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
						throw new RuntimeException(e);
					}
				}
			});

			latch.await();
			logger.info("completed");

		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} finally {
			try {
				httpclient.close();
			} catch (IOException ee) {
			}
		}
	}

	public void doScan(final String source, final String startS, final String endS) {
		logger.info("---------------Scan Start ----------------------------------");
		logger.info("source=" + source);
		logger.info("startS=" + startS);
		logger.info("endS=" + endS);

		Future<String> future = pool.submit(new Callable<String>() {
			@Override
			public String call() throws Exception {
				Utility.profiling(new Runnable() {
					@Override
					public void run() {
						try {
							HdfsHelper helper = new HdfsHelper();
							long startL = Long.valueOf(startS);
							long endL = Long.valueOf(endS);
							FileSystem hdfs = FileSystem.newInstance(helper.getConfig());
							Path srcPath = new Path(source);
							FSDataInputStream fileIn = hdfs.open(srcPath);
							scan(startL, endL, fileIn);
							fileIn.close();
							hdfs.close();
						} catch (IOException e) {
							e.printStackTrace();
							throw new RuntimeException(e);
						}
					}
				}, ">>>>>>>>> Scan on Data Node: ");
				return "";
			}
		});

		try {
			future.get();
		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (ExecutionException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
	}

	private void scan(long startL, long endL, FSDataInputStream fileIn) throws IOException {
		long length = endL - startL + 1;

		if (length <= 0) {
			throw new IllegalStateException("start position (" + startL + ") is greater than end position (" + endL
					+ ") !");
		}

		if (startL != 0) {
			fileIn.seek(startL);
		}

		long byteRead = 0L;
		byte[] buffer = new byte[(int) Math.min(length, ConstantsToBeExternalized.DEFAULT_HDFS_READ_BUFFER_SIZE)];
		while (byteRead < length) {
			int len = fileIn.read(buffer);
			if (len <= 0) {
				break;
			}
			byteRead += len;

			if (length - byteRead < ConstantsToBeExternalized.DEFAULT_HDFS_READ_BUFFER_SIZE) {
				buffer = new byte[(int) (length - byteRead)];
			}
		}
		logger.info("Total bytes scanned: " + byteRead);
	}
}
